﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using JetBlack.IO;
using JetBlack.SimpleTopicBus.Messages;

namespace JetBlack.SimpleTopicBus.Adapters
{
    /// <summary>
    /// A delegate called when data has been received.
    /// </summary>
    /// <param name="topic">The topic of the data.</param>
    /// <param name="data">The data itself.</param>
    /// <param name="isImage">When <c>true</c> the data is a full data set of the topic, otherwise it is a change.</param>
    public delegate void DataCallback(string topic, IDictionary<string, object> data, bool isImage);

    /// <summary>
    /// A delegate called when a subscription has been forwarded to this client by way of a notification request.
    /// </summary>
    /// <param name="clientId">The id of the client subscribing to the topic.</param>
    /// <param name="topic">The topic to which the client is subscribing.</param>
    /// <param name="isAdd">When <c>true</c> the client is adding a subscription, otherwiase the subscription is being removed.</param>
    public delegate void ForwardedSubscriptionCallback(int clientId, string topic, bool isAdd);

    /// <summary>
    /// A delegate called when the client closes.
    /// </summary>
    /// <param name="isAbnormal">If <c>true</c> the client has been closed abnormally.</param>
    public delegate void ClosedCallback(bool isAbnormal);

    /// <summary>
    /// A simple topic bus client.
    /// </summary>
    public class Client
    {
        /// <summary>
        /// The event raised when data has been received.
        /// </summary>
        public event DataCallback OnData;

        /// <summary>
        /// The event raised when a subscription request has been received on a topic on which this client has added a notification.
        /// </summary>
        public event ForwardedSubscriptionCallback OnForwardedSubscription;

        /// <summary>
        /// The event raised when a client is closed.
        /// </summary>
        public event ClosedCallback OnClosed;

        private TcpClient tcpClient;
        private bool isConnected;
        private NetworkReader reader;
        private NetworkWriter writer;
        private Thread thread;
        private NetworkStream networkStream;

        /// <summary>
        /// The default constructor.
        /// </summary>
        public Client()
        {
            isConnected = false;
            thread = new Thread(Dispatch);
        }

        /// <summary>
        /// A method to connect a client to a distributor.
        /// </summary>
        /// <param name="host">The host name of the distributor.</param>
        /// <param name="port">The port of the distributor.</param>
        /// <returns>On success <c>true</c> is returned; otherwise <c>false</c>.</returns>
        public bool Connect(string host, int port)
        {
            tcpClient = new TcpClient(host, port);
            networkStream = tcpClient.GetStream();

            reader = new NetworkReader(networkStream);
            writer = new NetworkWriter(networkStream);
            thread.Name = string.Format("interactor/{0}", tcpClient.Client.RemoteEndPoint);
            isConnected = true;
            thread.Start();

            return true;
        }

        /// <summary>
        /// The method to close a client.
        /// </summary>
        public void Close()
        {
            IsConnected = false;
            tcpClient.Close();
            thread.Join();
        }

        /// <summary>
        /// This method adds a subscription to a topic. When data is published on this topic an event will be raised on this client.
        /// </summary>
        /// <param name="topic">The topic of interest.</param>
        public void AddSubscription(string topic)
        {
            Write(new SubscriptionRequest(topic, true));
        }

        /// <summary>
        /// This method removes a previously registered subscription.
        /// </summary>
        /// <param name="topic">The unwanted topic.</param>
        public void RemoveSubscription(string topic)
        {
            Write(new SubscriptionRequest(topic, false));
        }

        /// <summary>
        /// Publish a full data set for this topic to a client.
        /// </summary>
        /// <param name="clientId">The destination client id. This will typically be known from a forwarded subscription through the notification mechanism.</param>
        /// <param name="topic">The topic.</param>
        /// <param name="isImage">If true the data is considered to be a full image of the data; otherwise it contains only changes.</param>
        /// <param name="data">The data.</param>
        public void Send(int clientId, string topic, bool isImage, IDictionary<string, object> data)
        {
            Write(new ClientMessage(clientId, topic, isImage, data));
        }

        /// <summary>
        /// Publish a change on a topic to all subscribed clients.
        /// </summary>
        /// <param name="topic">The topic.</param>
        /// <param name="isImage">If true the data is considered to be a full image of the data; otherwise it contains only changes.</param>
        /// <param name="data">The changed data.</param>
        public void Publish(string topic, bool isImage, IDictionary<string, object> data)
        {
            Write(new SubscriberMessage(topic, isImage, data));
        }

        /// <summary>
        /// Request notifications for clients that subscribe to topics which match the pattern.
        /// </summary>
        /// <param name="topicPattern">The topic pattern to match against subscription requests.</param>
        virtual public void AddNotification(string topicPattern)
        {
            Write(new NotificationRequest(topicPattern, true));
        }

        /// <summary>
        /// Remove notifications to a previously requested pattern.
        /// </summary>
        /// <param name="topicPattern">The topic pattern previously registered.</param>
        virtual public void RemoveNotification(string topicPattern)
        {
            Write(new NotificationRequest(topicPattern, false));
        }

        /// <summary>
        /// The connection state of the client.
        /// </summary>
        public bool IsConnected
        {
            get { lock (this) return isConnected; }
            set { lock (this) isConnected = value; }
        }

        /// <summary>
        /// The client socket end point.
        /// </summary>
        public IPEndPoint LocalEndPoint
        {
            get { return tcpClient.Client.LocalEndPoint as IPEndPoint; }
        }

        /// <summary>
        /// The distrubutor socket end point.
        /// </summary>
        public IPEndPoint RemoteEndPoint
        {
            get { return tcpClient.Client.RemoteEndPoint as IPEndPoint; }
        }

        private void Dispatch()
        {
            bool abnormalDisconnect = true;

            while (IsConnected)
            {
                try
                {
                    Message message = Message.Read(reader);

                    switch (message.MessageType)
                    {
                        case MessageType.SubscriberMessage:
                            RaiseOnData(message as SubscriberMessage);
                            break;
                        case MessageType.ClientMessage:
                            RaiseOnData(message as ClientMessage);
                            break;
                        case MessageType.ForwardedSubscriptionRequest:
                            RaiseOnForwardedSubscriptionRequest(message as ForwardedSubscriptionRequest);
                            break;
                        default:
                            throw new ArgumentException("invalid message type");
                    }
                }
                catch (System.IO.EndOfStreamException)
                {
                    IsConnected = false;
                    abnormalDisconnect = false;
                    continue;
                }
                catch (IOException ex)
                {
                    if (ex.InnerException != null && ex.InnerException is SocketException)
                    {
                        SocketException socketException = (SocketException)ex.InnerException;
                        if (socketException.ErrorCode == 10060)
                            continue;
                    }

                    if (IsConnected)
                    {
                        IsConnected = false;
                        abnormalDisconnect = true;
                    }
                    else
                        abnormalDisconnect = false;
                    continue;
                }
                catch (ObjectDisposedException)
                {
                    if (IsConnected)
                    {
                        IsConnected = false;
                        abnormalDisconnect = true;
                    }
                    else
                        abnormalDisconnect = false;

                    continue;
                }
            }

            if (OnClosed != null)
                OnClosed(abnormalDisconnect);
        }

        private void Write(Message message)
        {
            lock (writer)
                message.Write(writer);
        }

        private void RaiseOnForwardedSubscriptionRequest(ForwardedSubscriptionRequest message)
        {
            if (OnForwardedSubscription != null)
                OnForwardedSubscription(message.ClientId, message.Topic, message.IsAdd);
        }

        private void RaiseOnData(SubscriberMessage message)
        {
            RaiseOnData(message.Topic, message.Data, false);
        }

        private void RaiseOnData(ClientMessage message)
        {
            RaiseOnData(message.Topic, message.Data, true);
        }

        private void RaiseOnData(string topic, IDictionary<string, object> data, bool isImage)
        {
            if (OnData != null)
                OnData(topic, data, isImage);
        }
    }
}
